/*
   Copyright 2009 Abdulla Abdurakhmanov (abdulla@latestbit.com)
   Original sources are available at www.latestbit.com

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

package org.latestbit.smartobjects.core;


import java.io.IOException;

import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;

import org.latestbit.smartobjects.IServerSkeleton;
import org.latestbit.smartobjects.ISmartBus;
import org.latestbit.smartobjects.coders.StdNetworkCommand;
import org.latestbit.smartobjects.coders.StdPacketFormer;
import org.latestbit.smartobjects.exceptions.*;
import org.latestbit.smartobjects.io.ResizableByteBuffer;
import org.latestbit.smartobjects.network.*;

public class CommandProcessor implements ICommandProcessor {	
	
	protected Map<String, ICommandProcessorHandler> commandHandlersByDestinationFilter = new HashMap<String, ICommandProcessorHandler> ();
	protected Map<String, ICommandProcessorHandler> commandHandlersByInstanceFilter = new HashMap<String, ICommandProcessorHandler> ();

	private class AsyncCallTimerQueueItem extends TimerTask {
		private Object userObject;
		private INetworkCommand request;
		private ICommandProcessorHandler handler;

		public void setUserObject(Object userObject) {
			this.userObject = userObject;
		}

		public Object getUserObject() {
			return userObject;
		}

		public void setRequest(INetworkCommand command) {
			this.request = command;
		}

		public INetworkCommand getRequest() {
			return request;
		}

		public void setHandler(ICommandProcessorHandler handler) {
			this.handler = handler;
		}

		public ICommandProcessorHandler getHandler() {
			return handler;
		}

		@Override
		public void run() {
			boolean isRemoved = false;
			synchronized(asyncCalls) {
				isRemoved = asyncCalls.remove(this.getRequest().getMessageId()) != null;
			}
			if(isRemoved) {
				if ( handler!=null ) {
					handler.onResponseTimeout ( request, userObject );
				}
				else {
					if(userObject instanceof SyncCallContext) {
						SyncCallContext context = (SyncCallContext)userObject; 
						// It's a sync request
						synchronized(context) {
							context.syncCallResponse = null;
							context.notifyAll();
						}
					}
				}
			}
		}
	}
	
	private static class SyncCallContext { 
		public INetworkCommand syncCallResponse = null;	
	}
	
	private static class TlsSyncCallContext extends ThreadLocal<SyncCallContext> {
		public SyncCallContext initialValue() {
			return new SyncCallContext();
		}
	}
	
	private Timer asyncCallTimer = new Timer( "CommandProcessorTimer["+this+"]", true );
	
	private static final int ASYNC_CALL_TIMER_MAX_PURGE_COUNTER = 100;	
	private int asyncCallTimerPurgeCounter = 0;
	
	private Map< Integer,  AsyncCallTimerQueueItem> asyncCalls = 
		new HashMap < Integer, AsyncCallTimerQueueItem > ();
	
	private ISmartBus bus;
		
	protected LinkedList<ResizableByteBuffer> requestBuffersCache = new LinkedList<ResizableByteBuffer>();
	private ICommandProcessorHandler defaultProcessorHandler = new DefaultCommandProcessorHandler(this);
	private TlsSyncCallContext tlsSyncCallContext = new TlsSyncCallContext();
	
	public CommandProcessor(ISmartBus bus) {
		this.bus = bus;		
	}
	
	@Override
	public void destroy() {
		if(this.asyncCallTimer!=null) {
			this.asyncCallTimer.cancel();
			this.asyncCallTimer=null;
		}
		synchronized(asyncCalls) {			
			asyncCalls.clear();
		}
		synchronized(commandHandlersByDestinationFilter) {
			commandHandlersByDestinationFilter.clear();
			commandHandlersByInstanceFilter.clear();
		}		
	}
	
	@Override
	public void finalize() {
		destroy();
	}
	
	@Override
	public void addRequestHandler(ICommandProcessorHandler handler,
			String destinationFilter) {
		addRequestHandler ( handler, destinationFilter, null );
	}

	@Override
	public void addRequestHandler(ICommandProcessorHandler handler,
			String destinationFilter, String instanceIdFilter) {
		synchronized(commandHandlersByDestinationFilter) {
			ICommandProcessorHandler alreadyBounded = null;
			if(instanceIdFilter == null) {
				alreadyBounded = commandHandlersByDestinationFilter.get(destinationFilter);
				if(alreadyBounded!=null && alreadyBounded !=handler)
					throw new InternalError("Filter "+destinationFilter+"is already bound to another command processor handler");
				if(alreadyBounded == null)
					commandHandlersByDestinationFilter.put(destinationFilter, handler);
			}
			else {
				alreadyBounded = commandHandlersByInstanceFilter.get(instanceIdFilter);
				if(alreadyBounded !=null && alreadyBounded !=handler)
					throw new InternalError("Filter "+instanceIdFilter+"is already bound to another command processor handler");
				if(alreadyBounded == null)
					commandHandlersByInstanceFilter.put(instanceIdFilter, handler);
			}
		}
	}
	
	@Override
	public void addRequestHandler(ICommandProcessorHandler handler, IServerSkeleton smartObject) {
		addRequestHandler ( handler, smartObject.getInterfaceId(), smartObject.getInstanceId() );
		if ( smartObject.getSupportedInterfaces()!=null ) {
			for ( String supportedInterface : smartObject.getSupportedInterfaces() ) {
				addRequestHandler ( handler, supportedInterface );
			}
		}
			
	}	
	
	@Override
	public void delRequestHandler(ICommandProcessorHandler handler) {
		synchronized(commandHandlersByDestinationFilter) {
			Iterator<ICommandProcessorHandler> it = commandHandlersByDestinationFilter.values().iterator();
			while(it.hasNext()) {
				if ( it.next() == handler ) {
					it.remove();
					it = commandHandlersByDestinationFilter.values().iterator();
				}
			}
			
			it = commandHandlersByInstanceFilter.values().iterator();
			while(it.hasNext()) {
				if ( it.next() == handler ) {
					it.remove();
					it = commandHandlersByInstanceFilter.values().iterator();
				}
			}			
		}
	}
	
	@Override
	public void sendAsyncRequest(
			ConnectionHandler connectionHandler,
			INetworkCommand request,
			ICommandProcessorHandler handler, 
			Object userObject, 
			int timeoutInSec) throws IOException, BufferUnderflowException {
		
		if(!request.isRequest())
			throw new InternalError("Internal error: Unable to send response as request!");
		if(!request.isSerialized())
			throw new InternalError("Internal error: Request is not serialized before to send");
		
		AsyncCallTimerQueueItem item = new AsyncCallTimerQueueItem();
		item.setRequest(request);
		item.setHandler(handler);
		item.setUserObject(userObject);		
		
		synchronized(asyncCalls) {
			asyncCalls.put(request.getMessageId(), item );
		}
		
		// Date tm = new Date();
		// tm.setTime(tm.getTime()+timeoutInSec*1000);		
		// item.setTimerId( asyncCallTimer.addNotification(null,null, request.getMessageId(), tm) );
		asyncCallTimer.schedule( item , timeoutInSec*1000 );
		
		try {
			connectionHandler.getConnection().send(request.getBuffer().getBuffer());				
		}
		catch (Exception e) {
			item.cancel();
			synchronized(asyncCalls) {
				asyncCalls.remove(request.getMessageId());				
			}						
			throw new IOException(e.toString());
		}
	}
	
	
	@Override
	public void onReceived ( ConnectionHandler connectionHandler, ByteBuffer buffer ) {		
		
		StdNetworkCommand command = new StdNetworkCommand(buffer, getBus().createDataTypesCoder());
		try {
			command.deserialize();
			if(command.isRequest()) {
				onReceivedRequest(connectionHandler, command);
			}
			else {				
				onReceivedResponse(connectionHandler, command);
			}

		} catch (BufferUnderflowException ex) {						
			ex.printStackTrace();
			onReceiveError(connectionHandler, ex.toString());
		} catch (SmartBusSystemException ex) {			
			ex.printStackTrace();
			onReceiveError ( connectionHandler, ex.toString() );
		}
	}	

	protected void onReceivedResponse(ConnectionHandler connectionHandler, INetworkCommand response) {
		AsyncCallTimerQueueItem item = null;
		synchronized(asyncCalls) {			
			item = asyncCalls.get(response.getMessageId());
			if(item!=null) {
				asyncCalls.remove(response.getMessageId());
			}
		}
		
		if(item!=null) {
			item.cancel();
			if(++asyncCallTimerPurgeCounter > ASYNC_CALL_TIMER_MAX_PURGE_COUNTER) {
				asyncCallTimerPurgeCounter = 0;
				Timer timer = asyncCallTimer;
				if(timer!=null) {
					timer.purge();
				}
			}
			
			reuseRequestBuffer ( item.getRequest(), response );
			if(item.getUserObject() instanceof SyncCallContext) {
				SyncCallContext context = (SyncCallContext)item.getUserObject(); 
				// It's a sync request
				synchronized(context) {
					context.syncCallResponse = response;
					context.notifyAll();
				}
			}
			else {
				item.getHandler().onResponseReceived(connectionHandler, item.getRequest(), response, item.getUserObject());
			}
		}
		else {			
			// It's maybe a sync response (else it's timeout or error response)
		}
		
	}
	
	protected void reuseRequestBuffer (INetworkCommand request, INetworkCommand response) {
		// Reuse exist buffer instead create a new buffer for the response
		ResizableByteBuffer rszBuffer = request.getBuffer();
		request.setBuffer(null);
		rszBuffer.getBuffer().clear();
		rszBuffer.put( response.getBuffer().getBuffer() );
		rszBuffer.getBuffer().flip();
		response.setBuffer(rszBuffer);		
	}

	@Override
	public INetworkCommand sendSyncRequest(ConnectionHandler connectionHandler, INetworkCommand request, int timeoutInSec) throws IOException, TimeoutException, InterruptedException {
		if(!request.isRequest())
			throw new InternalError("Internal error: Unable to send response as request!");		
		if(!request.isSerialized())
			throw new InternalError("Internal error: Request is not serialized before to send");

		INetworkCommand response = null;
		SyncCallContext context = tlsSyncCallContext.get();
		synchronized(context) {		
			sendAsyncRequest(connectionHandler, request, null, context, timeoutInSec);
			context.wait();
			response = context.syncCallResponse;
			if(response == null)
				throw new TimeoutException();  
			context.syncCallResponse = null;		
		}
		return response;
	}

	
	@Override
	public void onReceiveError(ConnectionHandler handler, String errorMessage) {}

	@Override
	public void onWriteError(ConnectionHandler handler, String errorMessage) {}

	
	protected void onReceivedRequest( ConnectionHandler connectionHandler, INetworkCommand request) {
		ByteBuffer buffer = request.getBuffer().getBuffer();
		ResizableByteBuffer copyOfBuffer = null; 
		synchronized ( requestBuffersCache ) {
			if(requestBuffersCache.isEmpty()) {
				int initSize = buffer.limit() - buffer.position();
				if(initSize < 256)
					initSize = 256;
				copyOfBuffer = new ResizableByteBuffer( false, initSize, StdPacketFormer.MAX_PACKET_SIZE, 1024 );
			}
			else {				
				copyOfBuffer = requestBuffersCache.poll();
				copyOfBuffer.getBuffer().clear();
			}			
		}
		
		copyOfBuffer.put( buffer );
		copyOfBuffer.getBuffer().flip();
		request.setBuffer( copyOfBuffer );
		ICommandProcessorHandler processorHandler = null;
		synchronized(commandHandlersByDestinationFilter) {
			processorHandler = commandHandlersByInstanceFilter.get(request.getInstanceId());
			if(processorHandler == null)
				processorHandler =  commandHandlersByDestinationFilter.get(request.getDestination());
		}

		if(processorHandler!=null) {
			if(processorHandler.isCommandIdSupported(connectionHandler, request))
				processorHandler.onRequestReceived(connectionHandler, request);
			else
				processorHandler = null;
		}
		
		if(processorHandler == null)
			defaultProcessorHandler.onRequestReceived(connectionHandler, request);
		
	}
	
	@Override
	public void sendResponse(ConnectionHandler handler, INetworkCommand response) throws IOException {
		if(!response.isResponse())
			throw new InternalError("Internal error: Unable to send request as response!");			
		if(!response.isSerialized())
			throw new InternalError("Internal error: request is not serialized before to send");
		try {
			handler.getConnection().send( response.getBuffer().getBuffer() );
		}
		finally {
			synchronized( requestBuffersCache ) {
				requestBuffersCache.addLast( response.getBuffer() );				
				response.setBuffer(null);
			}
		}
	}

	public ISmartBus getBus() {
		return bus;
	}

	@Override
	public int getAsyncCallsQueueSize() {
		synchronized(asyncCalls) {			
			return asyncCalls.size();			
		}
	}

}
